package com.credithc.signature.core.rabbitmq;

import com.credithc.signature.common.ro.core.CoreRequet;
import com.credithc.signature.common.utils.UUIDUtil;
import com.credithc.signature.core.configure.rabbitmq.CoreRabbitConfig;
import com.credithc.signature.core.constant.Constant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 描述：生产者工具类<br/>
 * 作者：sunyang
 * 创建：2019/5/28 11:05
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Component(Constant.RABBITSENDER_NAME)
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class CoreRabbitSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    CoreRabbitConfig coreRabbitConfig;

    @Resource(name = CoreRabbitConfig.RABBITTEMPLATE_NAME)
    RabbitTemplate rabbitTemplate;

    final String id = UUIDUtil.getUUID();

    private volatile boolean failCallBack = false;
    private volatile boolean occurCallBack = false;

    /**
     * 同步发送消息
     *
     * @param message
     * @param vipStatus
     * @return
     */
    public boolean sendSync(Object message, Integer vipStatus) {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        boolean isVip = vipStatus == 0;
        String vip = "[普通队列]";

        try {
            CorrelationData correlationData = new CorrelationData(id);
            if (isVip) {
                rabbitTemplate.convertAndSend(coreRabbitConfig.getCallBackExchangeName(), coreRabbitConfig.getCallBackQueueName(), message,
                        messageNoneDurablePostProcessor, correlationData);
            } else {
                vip = "[VIP队列]";
                rabbitTemplate.convertAndSend(coreRabbitConfig.getVipCallBackExchangeName(), coreRabbitConfig.getVipCallBackQueueName(), message,
                        messageNoneDurablePostProcessor, correlationData);
            }

            if (!occurCallBack) {
                synchronized (this) {
                    this.wait(coreRabbitConfig.getSendTimeOut());
                }
            }
            if (!occurCallBack) {
                return false;
            }
            return !failCallBack;
        } catch (Exception e) {
            log.error("发送{}" + vip + "数据失败", message, e);
        }
        return false;
    }

    private final MessagePostProcessor messageNoneDurablePostProcessor = new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(final Message message) throws AmqpException {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        }
    };

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack && !failCallBack) {
            failCallBack = false;
        } else {
            failCallBack = true;
        }
        occurCallBack = true;
        synchronized (this) {
            this.notifyAll();
        }
        if (!failCallBack) {
            log.debug("成功发送消息－:requestSn={},", correlationData.getId());
        } else {
            log.error("发送消息:requestSn={},失败:cause{}", correlationData.getId(), cause);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("replyCode={},replyText={},exchange={},routingKey={},消息转换失败:{},", replyCode, replyText, exchange, routingKey,
                message.toString());
        failCallBack = true;
    }

    /**
     * 发送延迟消息
     *
     * @param exchange
     * @param routerKey
     * @param coreRequet
     * @param delayTimes
     */
    public void sendDelayMessage(String exchange, String routerKey, CoreRequet coreRequet, final long delayTimes) {
        try {
            // 执行发送消息到指定队列
            rabbitTemplate.convertAndSend(exchange, routerKey, coreRequet, message -> {
                // 设置延迟毫秒值
                message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                return message;
            });
            log.info("放入Core延迟队列成功...... requestSN:" + coreRequet.getRequestSn());
        } catch (Exception e) {
            log.error("放入Core延迟队列失败...... requestSN:" + coreRequet.getRequestSn(), e);
        }
    }
}
